01 基于 Langchain 开发一个多任务问答助手

总结

至此，我们已经成功构建了一个基于 LangChain 的多任务问答助手，它具备了以下核心功能：

• 多任务处理：通过 WeatherTool 和 NewsTool 实现了天气查询和新闻查询。
• 工具使用透明化：在回答中可以提示使用了哪些工具（通过解析 Agent 的输出）。
• 历史对话上下文：通过 RunnableWithMessageHistory 实现了对话记忆，让助手能够理解上下文。
• 缓存结果：通过 set_llm_cache(InMemoryCache()) 优化了 LLM 的重复调用。
• 基础错误处理：在工具内部和外部调用层都考虑了错误情况。

这个助手为你提供了一个强大的起点，你可以根据需要添加更多的工具（例如，日历、计算器、翻译等），替换为真实的 API，并部署到更健壮的环境中（例如，使用 Redis 作为会话存储和缓存）。

后记

回头看下，这个需求是实时的，输出是确定的。用 LLM并不是好的选择

这里提出的是一个非常重要且敏锐的观点：对于实时且输出确定的需求，直接使用 LLM 可能不是最佳选择。 我非常赞同你的看法。

LLM（大型语言模型）虽然强大，但在以下情况下确实存在局限性：

• 实时性要求高： LLM 的推理通常需要一定时间，特别是对于复杂的链式调用或高并发场景，这可能无法满足严格的实时性要求。
• 输出确定性强： LLM 本质上是概率模型，即使给定相同的输入，也可能产生略微不同的输出。对于需要精确、无歧义且可重复结果的场景（例如，获取确切的天气数据、股票价格、精确的计算结果），LLM 的这种不确定性是一个挑战。它们可能会“幻觉”出不存在的数据或错误地解释工具输出。
• 成本考虑： 每次调用 LLM 都会产生费用，对于高频的确定性查询，这可能是一笔不小的开销。
• 资源消耗： 运行 LLM 需要大量的计算资源。

更好的解决方案：结合传统编程与 LLM 的优势

对于你提出的“实时且输出确定”的需求，更优的方案是将 LLM 作为“意图识别和工具编排”的层，而将具体的、确定性的数据获取和处理任务交给传统的、精确的编程逻辑。

以下是一些建议的优化方向：

1. 明确意图识别层：
   • LLM 的核心作用： 让 LLM 主要负责理解用户输入的意图（例如：“查询天气”、“查找新闻”、“执行计算”）以及提取必要的参数（例如：城市名称、新闻关键词）。
   • 优点： LLM 在自然语言理解方面表现出色，能够灵活地处理用户不同的表达方式。
2. 确定性工具执行层：
   • 核心编程逻辑： 一旦 LLM 识别出意图和参数，就由传统的、预先编写好的函数或服务来执行具体的任务。
   • 示例：
     • 天气查询： 直接调用天气 API，并使用标准 JSON 解析和错误处理。
     • 新闻查询： 直接调用新闻 API，对结果进行格式化。
     • 其他确定性任务： 数据库查询、复杂的计算、系统命令执行等。
   • 优点： 快速、准确、可预测，且成本可控。
3. 结果返回与封装：
   • 工具直接返回结构化数据： 工具的输出应该是清晰、结构化的，而不是一段自由文本。
   • LLM 封装最终答案（可选）： 如果需要更人性化或更复杂的自然语言回复，LLM 可以用来将工具返回的结构化数据封装成自然语言答案。但这应该是最后一步，并且在简单的场景下甚至可以省略，直接由程序生成答案。

对现有框架的改进思路

基于我们之前构建的 LangChain 助手，可以这样优化：

• 工具实现更健壮： WeatherTool 和 NewsTool 的 _run 方法内部，应该彻底移除模拟数据，完全依赖真实的 API 调用，并包含详尽的错误处理（网络错误、API 密钥错误、数据解析错误等）。确保这些工具本身是确定且可靠的。
• 减少 LLM 的“自由发挥”：
  • 精炼 Prompt： 进一步优化 LLM 的 Prompt，明确指示它仅仅作为“路由员”和“参数提取器”。告诉它如果工具返回错误，就直接报告错误，而不是尝试“猜测”或“创造”答案。
  • 强制工具调用： 在某些情况下，可以尝试引导 LLM 更倾向于调用工具，或者在无法确定意图时，请求用户提供更多信息。
• 性能监控与优化： 对于实时性要求高的部分，监控工具的调用延迟，并考虑异步调用或批处理。

总结

这个观点非常正确，它引导我们走向一个更实用、更高效的架构：LLM 负责“理解”和“决策”，而传统代码负责“执行”和“确定性数据处理”。这是一种将两者优势结合的强大模式，尤其适用于多任务问答助手场景。

讨论

也就是LLM解决只是人机交互使用自然语言的界面问题，但是这个还是觉得不用 LLM更加让人信服

你再次提出了一个非常深刻且关键的问题：LLM 的真正价值是否仅限于自然语言交互界面，以及在某些场景下，完全不使用 LLM 是否更具说服力？

你的直觉是正确的。对于许多具有明确规则、确定性输入和输出，以及高实时性要求的系统来说，引入 LLM 确实可能带来不必要的复杂性、不确定性、延迟和成本，从而降低系统的“说服力”和可靠性。

LLM 的核心价值与局限性再审视

让我详细阐述一下我的看法：

1. LLM 的核心优势：非结构化数据的理解与生成

   • 自然语言理解 (NLU)：LLM 最擅长的是理解人类的自然语言输入，包括其意图、情感、上下文和细微差别。这是传统规则或正则表达式方法难以比拟的。
   • 自然语言生成 (NLG)：LLM 可以生成流畅、连贯、符合语境的自然语言文本，使得机器的回复更具“人情味”和可读性。
   • 知识推理与通用性：在没有明确规则或预定义路径的情况下，LLM 可以在其训练数据范围内进行一定程度的推理和联想，处理开放域问题。

2. LLM 在确定性任务中的局限性

   • 非确定性输出：这是最大的问题。你问它“2+2等于几”，它大部分时候会说“4”，但偶尔也可能“幻觉”出其他答案。这对于需要精确计算、数据检索的任务是致命的。
   • 推理链过长导致的不稳定性：当 LLM 需要进行多步思考（例如：理解意图 -> 提取参数 -> 选择工具 -> 调用工具 -> 解析结果 -> 生成回复）时，每一步都可能引入误差，导致最终结果不可靠。
   • 高延迟与成本：调用 LLM 通常比执行本地代码或查询简单数据库慢得多，而且每次调用都会产生费用。对于高频次、实时性强的应用，这是不可接受的。
   • 可解释性差：你很难完全理解 LLM 是如何得出某个结论或选择某个工具的，这使得调试和维护变得困难。
   • 训练数据偏差与时效性：LLM 的知识停留在其训练数据的时间点，对于实时变化的最新信息（如今天的天气、最新的新闻事件）是无能为力的，必须依赖外部工具。

何时“不用 LLM 更让人信服”？

当你的需求满足以下条件时，完全舍弃 LLM，采用传统的编程方法，会是更“说服力”的选择：

1. 明确的规则和逻辑：你的系统可以通过一系列 if-else、模式匹配、有限状态机等方式来明确定义。
   • 例子： 一个根据关键词在固定数据库中查找产品的系统，一个根据固定命令执行操作的系统。
2. 确定的输入和输出：输入格式相对固定，你期望的输出是精确、无二义的。
   • 例子： API 网关，数据转换工具，命令行工具。
3. 高性能和低延迟要求：系统必须在毫秒级别响应。
4. 成本敏感：每次操作都希望成本尽可能低。
5. 严格的可靠性和可解释性：系统必须 100% 可靠，并且其行为必须是可审计、可预测的。
6. 简单的自然语言交互：如果用户只需要使用非常有限、固定或结构化的自然语言短语（例如“查天气 北京”、“新闻 科技”），那么传统的自然语言处理（NLP）技术，如关键词匹配、N-gram 模型或基于规则的意图识别，就足够了，而且更高效、更稳定。

你的多任务问答助手的实际考量

回到你的多任务问答助手：

• “帮我查一下今天的天气”：
  • 如果只要求精确识别“天气”和“城市”，传统的 NLP 意图识别模块（如 Rasa NLU、Dialogflow Essentials，或者自己基于正则和词典的简单实现）完全可以胜任，它们更确定、更快。
  • 然后直接调用天气 API。
• “最近的新闻有哪些？”：
  • 同样，识别“新闻”意图，提取“关键词”，然后调用新闻 API。

在这种场景下，LLM 的角色被大大削弱，甚至可以被完全取代。它主要用来处理那些用户输入高度多样化、模糊或需要通用知识才能理解的场景。

我的建议：渐进式优化或直接采用传统方案

1. 对于确定性强、实时性高的部分，优先考虑传统编程：

   • 意图识别与参数提取： 对于常用的、预设的问题类型，使用规则匹配、关键词提取或小型、专门训练的模型（如简单的机器学习分类器）来识别意图和参数。这会比 LLM 更快、更稳定。
   • 工具调用与结果处理： 这部分应始终是传统的、强类型、强错误处理的函数或服务。

2. 如果 LLM 的“自然语言界面”是核心价值：

   • 如果用户希望以非常随意、多样化的方式提问，并且你愿意接受一定的不确定性、延迟和成本来换取这种交互的流畅性，那么 LLM 作为前端的自然语言接口仍然有其价值。它将用户的“模糊语言”转化为“清晰指令”，然后将指令传递给后端的确切工具。
   • 但在这种情况下，需要通过非常精心设计的 Prompt Engineering 和 RAG（检索增强生成） 来约束 LLM 的行为，让它知道自己的边界，不要“幻觉”，而是专注于调用工具和格式化工具的输出。

对于很多业务系统，可靠性、确定性、性能和成本是首要考虑的因素。LLM 虽然是强大的通用工具，但并非所有问题的银弹。明智地选择何时使用、何时不使用 LLM，并将其与传统编程的优势结合，才能构建出最强大、最可信服的解决方案。

所以，我们是否应该考虑调整一下方向，为你的助手实现一个更轻量级、更确定性的意图识别层，而不是完全依赖 LLM 来进行工具选择？或者，你更倾向于在现有 LLM 框架下，最大限度地提高其确定性和鲁棒性？

完整代码（不使用 LLM ）

需求回顾： * 目标： 开发一个多任务问答助手。 * 输入： 用户自然语言问题（如：“帮我查一下今天的天气”、“最近的新闻有哪些？”）。 * 输出： 准确答案 + 使用了哪些工具。 * 扩展项： 支持历史对话上下文、缓存结果、错误处理。 * 核心痛点（你提出的）： 对于实时、确定性输出的需求，LLM 可能不是最佳选择，更倾向于直接、确定的代码实现。

编写一个基于传统编程逻辑的多任务问答助手代码。这个版本将不依赖于 LangChain Agent 和 LLM 来进行意图识别和工具选择，而是使用更直接、更确定的规则（如关键词匹配）。

这样，我们可以确保： * 实时性更高：没有 LLM 调用带来的延迟。 * 确定性更强：基于明确的规则进行匹配和处理。 * 成本更低：无需支付 LLM 调用费用。 * 可解释性更强：代码逻辑一目了然。

基于传统规则的多任务问答助手代码

我们将实现一个简单的基于关键词匹配的意图识别器，然后根据识别到的意图调用相应的工具函数。

1. 项目结构（建议）：

multi_task_assistant/
├── .env
├── main.py
├── tools.py
└── requirements.txt

2. requirements.txt 文件内容：

python-dotenv
requests

3. .env 文件内容 (请替换为你的真实 API 密钥，如果使用真实 API)：

# OPENWEATHERMAP_API_KEY="你的OpenWeatherMap API密钥"
# NEWS_API_KEY="你的NewsAPI密钥"

4. tools.py 文件内容 (定义具体的工具函数)：

import os
import requests
import datetime # 用于模拟新闻日期

# --- 模拟数据或真实API凭证 ---
# 如果你没有真实的API密钥，或者想快速测试，可以使用模拟数据
# MOCK_WEATHER_DATA 和 MOCK_NEWS_DATA
# 如果你想使用真实API，请确保在.env中配置了API密钥，并取消下面代码中的注释。

# OpenWeatherMap API配置 (真实的API调用)
# 你需要在OpenWeatherMap注册并获取API密钥
# OPENWEATHERMAP_API_KEY = os.getenv("OPENWEATHERMAP_API_KEY")
# BASE_WEATHER_URL = "https://api.openweathermap.org/data/2.5/weather"

# NewsAPI 配置 (真实的API调用)
# 你需要在NewsAPI注册并获取API密钥
# NEWS_API_KEY = os.getenv("NEWS_API_KEY")
# BASE_NEWS_URL = "https://newsapi.org/v2/everything"


# --- 模拟天气工具 ---
def get_weather_mock(city: str) -> str:
    """模拟获取指定城市的天气信息。"""
    mock_data = {
        "北京": {"temperature": 25, "condition": "晴朗", "humidity": 60, "source": "模拟天气数据"},
        "上海": {"temperature": 28, "condition": "多云", "humidity": 75, "source": "模拟天气数据"},
        "纽约": {"temperature": 20, "condition": "小雨", "humidity": 80, "source": "模拟天气数据"},
        "伦敦": {"temperature": 18, "condition": "阴天", "humidity": 90, "source": "模拟天气数据"},
    }
    weather_info = mock_data.get(city.lower()) # 转换为小写以匹配
    if weather_info:
        return f"{city}的天气是：{weather_info['condition']}，温度：{weather_info['temperature']}°C，湿度：{weather_info['humidity']}%。 (来源: {weather_info['source']})"
    else:
        return f"抱歉，我没有找到{city}的天气信息。 (来源: 模拟天气工具)"

# --- 真实天气工具 (使用OpenWeatherMap) ---
def get_weather_real(city: str) -> str:
    """通过OpenWeatherMap API获取指定城市的天气信息。"""
    OPENWEATHERMAP_API_KEY = os.getenv("OPENWEATHERMAP_API_KEY")
    BASE_WEATHER_URL = "https://api.openweathermap.org/data/2.5/weather"

    if not OPENWEATHERMAP_API_KEY:
        return "错误：OpenWeatherMap API 密钥未配置。"

    params = {
        "q": city,
        "appid": OPENWEATHERMAP_API_KEY,
        "units": "metric", # 摄氏度
        "lang": "zh_cn" # 中文描述
    }
    try:
        response = requests.get(BASE_WEATHER_URL, params=params, timeout=5)
        response.raise_for_status() # 对 4xx/5xx 响应抛出 HTTPError
        data = response.json()

        temp = data['main']['temp']
        condition = data['weather'][0]['description']
        humidity = data['main']['humidity']
        return f"{city}的天气是：{condition}，温度：{temp}°C，湿度：{humidity}%。 (来源: OpenWeatherMap)"
    except requests.exceptions.Timeout:
        return f"查询{city}天气超时，请稍后再试。 (来源: OpenWeatherMap)"
    except requests.exceptions.ConnectionError:
        return "网络连接错误，无法查询天气，请检查您的网络。 (来源: OpenWeatherMap)"
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 404:
            return f"抱歉，没有找到城市 '{city}' 的天气信息。 (来源: OpenWeatherMap)"
        return f"获取{city}天气失败：HTTP错误 {e.response.status_code} - {e.response.text} (来源: OpenWeatherMap)"
    except Exception as e:
        return f"查询{city}天气时发生未知错误：{e} (来源: OpenWeatherMap)"


# --- 模拟新闻工具 ---
def get_news_mock(query: str = None) -> str:
    """模拟获取新闻。"""
    mock_news = [
        {"title": "全球经济论坛在北京召开", "source": "新华社", "date": (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")},
        {"title": "人工智能技术取得突破性进展", "source": "科技日报", "date": (datetime.date.today() - datetime.timedelta(days=2)).strftime("%Y-%m-%d")},
        {"title": "某地发现稀有野生动物", "source": "自然周刊", "date": (datetime.date.today() - datetime.timedelta(days=3)).strftime("%Y-%m-%d")},
        {"title": "夏季旅游市场持续火爆", "source": "旅游指南", "date": (datetime.date.today() - datetime.timedelta(days=4)).strftime("%Y-%m-%d")},
    ]

    if query:
        filtered_news = [
            n for n in mock_news if query.lower() in n["title"].lower()
        ]
        if filtered_news:
            news_list = [
                f"标题：《{n['title']}》，来源：{n['source']}，日期：{n['date']}"
                for n in filtered_news
            ]
            return f"关于'{query}'的新闻：\n" + "\n".join(news_list) + " (来源: 模拟新闻数据)"
        else:
            return f"抱歉，没有找到关于'{query}'的新闻。 (来源: 模拟新闻工具)"
    else:
        news_list = [
            f"标题：《{n['title']}》，来源：{n['source']}，日期：{n['date']}"
            for n in mock_news
        ]
        return "最新热点新闻：\n" + "\n".join(news_list) + " (来源: 模拟新闻数据)"

# --- 真实新闻工具 (使用NewsAPI) ---
def get_news_real(query: str = None) -> str:
    """通过NewsAPI获取新闻。"""
    NEWS_API_KEY = os.getenv("NEWS_API_KEY")
    BASE_NEWS_URL = "https://newsapi.org/v2/everything"

    if not NEWS_API_KEY:
        return "错误：NewsAPI 密钥未配置。"

    params = {
        "q": query if query else "latest news", # 如果没有关键词，搜索"latest news"
        "apiKey": NEWS_API_KEY,
        "language": "zh", # 中文新闻
        "sortBy": "publishedAt",
        "pageSize": 3 # 只取3条
    }
    try:
        response = requests.get(BASE_NEWS_URL, params=params, timeout=5)
        response.raise_for_status()
        data = response.json()

        articles = data.get('articles', [])
        if articles:
            news_list = []
            for article in articles:
                title = article.get('title', '无标题')
                source_name = article.get('source', {}).get('name', '未知来源')
                published_at = article.get('publishedAt', '')[:10] # 只取日期部分
                news_list.append(f"标题：《{title}》，来源：{source_name}，日期：{published_at}")
            return "最新新闻：\n" + "\n".join(news_list) + " (来源: NewsAPI)"
        else:
            return f"抱歉，没有找到关于'{query}'的新闻。 (来源: NewsAPI)"
    except requests.exceptions.Timeout:
        return f"查询新闻超时，请稍后再试。 (来源: NewsAPI)"
    except requests.exceptions.ConnectionError:
        return "网络连接错误，无法查询新闻，请检查您的网络。 (来源: NewsAPI)"
    except requests.exceptions.HTTPError as e:
        return f"获取新闻失败：HTTP错误 {e.response.status_code} - {e.response.text} (来源: NewsAPI)"
    except Exception as e:
        return f"查询新闻时发生未知错误：{e} (来源: NewsAPI)"

# 选择使用模拟还是真实工具
# 如果你想使用真实API，请将这里的 USE_REAL_API 设置为 True，并确保 .env 配置正确
USE_REAL_API = False # 设置为 True 即可切换到真实API

WEATHER_FUNCTION = get_weather_real if USE_REAL_API else get_weather_mock
NEWS_FUNCTION = get_news_real if USE_REAL_API else get_news_mock

print(f"工具初始化完成，当前使用 {'真实API' if USE_REAL_API else '模拟数据'}。")

import os
from dotenv import load_dotenv
from collections import deque
import re
import json # 用于缓存序列化
import hashlib # 用于生成缓存key
import time # 用于缓存过期

# 从 tools.py 导入工具函数
from tools import WEATHER_FUNCTION, NEWS_FUNCTION

# 加载环境变量
load_dotenv()

# --- 配置 ---
MAX_CHAT_HISTORY = 5 # 存储最近5轮对话
CACHE_TTL_SECONDS = 300 # 缓存有效期5分钟

# --- 全局缓存 ---
# 简单的内存缓存，key: hash(query + context), value: (result, timestamp)
query_cache = {}

# --- 会话历史存储 ---
# 模拟会话存储，实际应用中可以是数据库或Redis
session_history_store = {}

class ChatSession:
    def __init__(self, session_id: str):
        self.session_id = session_id
        # deque 提供了高效的队列操作，适用于固定大小的历史记录
        self.history = deque(maxlen=MAX_CHAT_HISTORY) # 存储 (role, message)

    def add_message(self, role: str, message: str):
        self.history.append((role, message))

    def get_history_messages(self) -> list:
        return list(self.history)

def get_session(session_id: str) -> ChatSession:
    if session_id not in session_history_store:
        session_history_store[session_id] = ChatSession(session_id)
    return session_history_store[session_id]

# --- 意图识别和参数提取 ---
def parse_input_and_identify_intent(user_input: str, session_id: str) -> dict:
    """
    根据用户输入识别意图并提取参数。
    这里使用简单的关键词匹配和正则，可以根据需要扩展为更复杂的NLP模型。
    支持历史上下文来推断省略的参数。
    """
    input_lower = user_input.lower()
    session = get_session(session_id)
    history = session.get_history_messages()

    intent_data = {"intent": "unknown", "tool_used": "None", "params": {}}

    # 1. 天气查询
    weather_keywords = ["天气", "温度", "多少度", "冷不冷", "热不热", "下雨"]
    for keyword in weather_keywords:
        if keyword in input_lower:
            # 尝试从当前输入中提取城市
            city_match = re.search(r'(在|是|查|看|的|关于)?\s*([a-zA-Z\u4e00-\u9fa5]+?)(市|县|镇|区)?\s*(今天|明天|后天)?\s*(的)?\s*(天气|温度|多少度|冷不冷|热不热|下雨)', user_input)
            if city_match:
                city = city_match.group(2)
                intent_data["intent"] = "weather_query"
                intent_data["params"]["city"] = city
                break
            # 尝试从历史中获取城市（例如：“那上海呢？”）
            if "那" in input_lower and ("呢" in input_lower or "?" in input_lower):
                for role, msg in reversed(history):
                    if role == "ai_tool_response" and "天气" in msg:
                        prev_city_match = re.search(r'(.+?)的天气是', msg)
                        if prev_city_match:
                            prev_city = prev_city_match.group(1)
                            intent_data["intent"] = "weather_query"
                            intent_data["params"]["city"] = prev_city # 沿用上一个城市的意图
                            break
            if intent_data["intent"] == "unknown": # 如果没有明确城市，但有天气关键词
                intent_data["intent"] = "weather_query" # 标记为天气查询，但可能缺少城市

    # 2. 新闻查询
    news_keywords = ["新闻", "消息", "发生什么", "头条", "资讯"]
    for keyword in news_keywords:
        if keyword in input_lower:
            intent_data["intent"] = "news_query"
            # 尝试提取新闻关键词
            query_match = re.search(r'(关于|有关)?\s*([^，。？！]+?)(的)?\s*(新闻|消息|资讯)', user_input)
            if query_match and query_match.group(2):
                intent_data["params"]["query"] = query_match.group(2).strip()
            else:
                intent_data["params"]["query"] = None # 没有明确关键词则返回最新热点
            break

    # 3. 其他/通用问候
    greeting_keywords = ["你好", "您好", "在吗", "你是谁"]
    for keyword in greeting_keywords:
        if keyword in input_lower:
            intent_data["intent"] = "greeting"
            break

    return intent_data

# --- 缓存管理 ---
def get_cache_key(query: str, intent_data: dict) -> str:
    """生成缓存key，考虑查询和意图参数。"""
    data_str = json.dumps({"query": query, "intent": intent_data}, sort_keys=True)
    return hashlib.md5(data_str.encode('utf-8')).hexdigest()

def get_from_cache(key: str):
    """从缓存中获取数据，如果过期则移除。"""
    if key in query_cache:
        result, timestamp = query_cache[key]
        if time.time() - timestamp < CACHE_TTL_SECONDS:
            print(f"[缓存命中] Key: {key}")
            return result
        else:
            print(f"[缓存过期] Key: {key}")
            del query_cache[key]
    return None

def set_to_cache(key: str, result: str):
    """将数据存入缓存。"""
    query_cache[key] = (result, time.time())
    print(f"[缓存写入] Key: {key}")

# --- 主助手逻辑 ---
def multi_task_assistant(user_input: str, session_id: str = "default_session") -> str:
    session = get_session(session_id)
    session.add_message("human", user_input)

    # 1. 意图识别与参数提取
    intent_data = parse_input_and_identify_intent(user_input, session_id)
    intent = intent_data["intent"]
    params = intent_data["params"]

    response = ""
    tool_used = "None"

    # 尝试从缓存中获取
    cache_key = get_cache_key(user_input, intent_data)
    cached_response = get_from_cache(cache_key)
    if cached_response:
        response = cached_response
        tool_used = "Cache" # 标记为缓存命中
    else:
        # 2. 根据意图执行相应操作
        try:
            if intent == "weather_query":
                city = params.get("city")
                if city:
                    response = WEATHER_FUNCTION(city)
                    tool_used = WEATHER_FUNCTION.__name__
                else:
                    response = "请问您想查询哪个城市的天气呢？"
                    tool_used = "N/A (需要更多信息)"
            elif intent == "news_query":
                query = params.get("query")
                response = NEWS_FUNCTION(query)
                tool_used = NEWS_FUNCTION.__name__
            elif intent == "greeting":
                response = "你好！我是一个多任务问答助手，我可以帮你查询天气或新闻。请问有什么可以帮您的？"
                tool_used = "N/A (问候)"
            else:
                response = "抱歉，我不明白您的意思。我目前只能查询天气和新闻。请问您需要了解哪方面的信息？"
                tool_used = "N/A (未知意图)"
        except Exception as e:
            response = f"抱歉，在处理您的请求时发生了一个内部错误：{e}"
            tool_used = "Error"

        # 3. 缓存结果 (仅缓存成功且非交互式的结果)
        if tool_used not in ["Cache", "N/A (需要更多信息)", "Error"]:
            set_to_cache(cache_key, response)

    # 4. 记录AI响应到历史
    session.add_message("ai_tool_response", response)

    return f"{response} (使用了工具: {tool_used})"

# --- 命令行交互循环 ---
if __name__ == "__main__":
    print("欢迎使用多任务问答助手！")
    print("你可以问：'北京天气怎么样？' 或 '最近有什么新闻？' 或 '关于人工智能的新闻'")
    print("输入 '退出' 结束对话。")

    current_session_id = "user_session_1" # 可以根据实际情况动态生成

    while True:
        user_input = input(f"\n[{current_session_id}] 你: ").strip()
        if user_input.lower() == "退出":
            print("再见！")
            break

        assistant_response = multi_task_assistant(user_input, current_session_id)
        print(f"[{current_session_id}] 助手: {assistant_response}")